package cn.boat.router.consumer;

import org.apache.zookeeper.ZooKeeper;
import org.junit.Test;

import java.util.concurrent.CountDownLatch;

public class Demo {
    int threadNumber = 2;
    final CountDownLatch cdl = new CountDownLatch(threadNumber);//参数为线程个数
    ZooKeeper zooKeeperConnection = null;
    @Test
    public void test1()throws Exception {



        Thread threadOne = new Thread(()->{
            try {
                // 建立客户端与zookeeper服务器的连接
                ConsumerConnectZookeeper consumerConnectZookeeper = new ConsumerConnectZookeeper();
                zooKeeperConnection = consumerConnectZookeeper.connect();
        /*RegisterCenterImpl registerCenter = new RegisterCenterImpl(zooKeeperConnection);
        registerCenter.register("consumerGroup001","consumer002");
        SubscribeTopcisImpl subscribeTopcis = new SubscribeTopcisImpl();*/
                // consumer注册，不能重复注册 传递的参数（String  消费者名，String 消费者ID ）
                RegisterCenterImpl registerCenterMy = new RegisterCenterImpl(zooKeeperConnection);
                registerCenterMy.register("consumerGroup001","consumer002");
                // consumer订阅topic（Zookeeper zookeeper连接，String  消费者名，String 消费者ID，String topic，int 指定想要订阅主题的分区数量 ）
                // 返回值为consumer的元信息 如果该节点没有订阅topic，则返回值为空
                SubscribeTopcisImpl subscribeTopcis = new SubscribeTopcisImpl();
                System.out.println("consumer002开始监控。。。。。。。。。。。。");
                ConsumerMetaData consumerMetaData = subscribeTopcis.subscribeTopics(zooKeeperConnection, "consumerGroup001", "consumer002", "topic-1", 2);
                System.out.println("consumer002的cosumer元数据"+consumerMetaData);
                // 对zookeeper上的节点进行监控
                ConsumerWatcherImpl consumerWatcher = new ConsumerWatcherImpl(zooKeeperConnection);
                // 监控consumer的增加或者减少， 增加不做变动，有消费者减少，需要记录消费者之前消费过的分区以及主题，还有新的消费者
                consumerWatcher.watchConsumerId("consumerGroup001","consumer002");
                // 监控brokerid 的变化，重新获取consumer元数据
                // 监控topic的分区leader state节点的变化，重新获取consumer元数据
                consumerWatcher.watchIntrestingTopicsBrokerLeaderChange("consumerGroup001","consumer002","topic-1");

                Thread.sleep(400000);
                System.out.println("consumer002 is over");
                zooKeeperConnection.close();
                cdl.countDown();
            }catch (Exception e){
                e.printStackTrace();
            }
        });
        Thread threadTwo = new Thread(()->{
            try {

                // 建立客户端与zookeeper服务器的连接
                ConsumerConnectZookeeper consumerConnectZookeeper = new ConsumerConnectZookeeper();
                zooKeeperConnection = consumerConnectZookeeper.connect();
        /*RegisterCenterImpl registerCenter = new RegisterCenterImpl(zooKeeperConnection);
        registerCenter.register("consumerGroup001","consumer002");
        SubscribeTopcisImpl subscribeTopcis = new SubscribeTopcisImpl();*/
                // consumer注册，不能重复注册 传递的参数（String  消费者名，String 消费者ID ）
                RegisterCenterImpl registerCenterMy = new RegisterCenterImpl(zooKeeperConnection);
                registerCenterMy.register("consumerGroup001","consumer003");

                // consumer订阅topic（Zookeeper zookeeper连接，String  消费者名，String 消费者ID，String topic，int 指定想要订阅主题的分区数量 ）
                // 返回值为consumer的元信息 如果该节点没有订阅topic，则返回值为空
                SubscribeTopcisImpl subscribeTopcis = new SubscribeTopcisImpl();
                System.out.println("consumer003开始监控。。。。。。。。。。。。");
                ConsumerMetaData consumerMetaData = subscribeTopcis.subscribeTopics(zooKeeperConnection, "consumerGroup001", "consumer003", "topic-2", 2);
                // 对zookeeper上的节点进行监控
                ConsumerWatcherImpl consumerWatcher = new ConsumerWatcherImpl(zooKeeperConnection);
                // 监控consumer的增加或者减少， 增加不做变动，有消费者减少，需要记录消费者之前消费过的分区以及主题，还有新的消费者
                consumerWatcher.watchConsumerId("consumerGroup001","consumer003");
                // 监控brokerid 的变化，重新获取consumer元数据
                // 监控topic的分区leader state节点的变化，重新获取consumer元数据
                consumerWatcher.watchIntrestingTopicsBrokerLeaderChange("consumerGroup001","consumer003","topic-2");
                Thread.sleep(45000);
                System.out.println("consumer003 is over");
                zooKeeperConnection.close();
                cdl.countDown();
            }catch (Exception e){
                e.printStackTrace();
            }
        });
        threadOne.start();
        Thread.sleep(20000);
        threadTwo.start();
        try {
            cdl.await();//需要捕获异常，当其中线程数为0时这里才会继续运行
        }catch (InterruptedException e){
            e.printStackTrace();
        }

    }
}
